package com.example.demo.configer;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.example.demo.manager.publisher.CanalChangePublisher;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.List;

/**
 * @Description
 * @Author <a href="hedong.wu@baozun.com">wuhedong</a>
 * @Date 2020/5/26 16:11
 **/
@Service
@Slf4j
@AllArgsConstructor
public class CanalJob {


    private final CanalConfiguration canalConfiguration;

    private final CanalConnector canalConnector;

    private final CanalChangePublisher<?> canalChangePublisher;

    @Scheduled(cron = "0/1 * * * * ?")
    public void cron() {
        Message message = canalConnector.getWithoutAck(canalConfiguration.getBatchSize());
        long batchId = message.getId();
        int size = message.getEntries().size();
        if (batchId == -1 || size == 0) {
            return;
        }
        List<CanalEntry.Entry> entries = message.getEntries();
        for (CanalEntry.Entry entry : entries) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }
            canalChangePublisher.publish(entry);
        }
        canalConnector.ack(batchId);
    }

}
